package org.myorg.v4;

/*
 * Compare democratic model's PGA running time 
 *  
 */

import java.io.*;
import java.util.*;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapred.*;
import org.apache.hadoop.util.*;

import org.apache.hadoop.fs.FileSystem;
//import org.myorg.v3.DRecordReader;
//import org.myorg.v3.RWF3;

/*
 * Test serial and democratic standalone GA
 * 
 * compare the running time, including wall clock and iterator times
 * so, mapper's output is solving quality, wall clock and iterator times
 * 
 * experiment repeat 25, show the best. worst and mean time *  
 * 
 * mapper's number from 2 to 16
 * 
 */


public class Time extends Para4 {

	private static final Log LOG = LogFactory.getLog(Time.class);
	
	public static int ser = 0;

	public static class DPGAInputFormat extends 
	FileInputFormat<IntWritable, IntWritable> {
		
		public int begin = 0;
		
		@SuppressWarnings("deprecation")
		public InputSplit[] getSplits(JobConf conf, int numSplits) 
		throws IOException {
			
			
			FileStatus[] files = listStatus(conf);
		    // generate splits
		    ArrayList<FileSplit> splits = new ArrayList<FileSplit>(dmn);
		    
		    for (FileStatus file: files) {
		        Path path = file.getPath();
		        FileSystem fs = path.getFileSystem(conf);
		        long length = file.getLen();
		        long start = 0;
		        FSDataInputStream in = fs.open(path);
		        
		        
		        LOG.info("file length is " + length);
		       
		        //Create empty hosts array for zero length files
		        for (int i = 0; i < dmn; i++) {
		        	//long length = file.getLen();
		        	String line = in.readLine();
		        	int l = line.length();
		        	LOG.info("put start is " + start + " line is " + line);
		        	splits.add(new FileSplit(path, start, l, new String[0]));
		        	start = start + l + 1;
		        }
		        
		      }
		    
/*		    FileSplit[] ss = new FileSplit[splits.size()];
		    splits.toArray(ss);
		    
		      LOG.info("stanley Total # of splits: " + splits.size());
		      for (int t = 0; t < splits.size(); t++ ) {
		    	  LOG.info("start is " + splits.get(t).getStart());
		    	  
					
					final Path file = ss[t].getPath();
				    FileSystem fs = file.getFileSystem(conf);
				    final FSDataInputStream fileIn = fs.open(ss[t].getPath());
				    fileIn.seek(ss[t].getStart());
				    String line = fileIn.readLine();
				    //final ObjectInputStream object = new ObjectInputStream(fileIn);
				   //FPeopleWritable2 fp = new FPeopleWritable2(peopleNum/mmn,geneLength);
				   //fp.readFields(fileIn);
				    LOG.info("XXX: " + ss[t].getStart() + "line is " + line);
		    	  
		      }*/
		      
		      return splits.toArray(new FileSplit[splits.size()]);



		}
		
		public RecordReader<IntWritable,IntWritable> getRecordReader(
				InputSplit split, JobConf conf, Reporter reporter)
				throws IOException {
			FileSplit fsplit = (FileSplit)split;
			final Path file = fsplit.getPath();
		    // open the file and seek to the start of the split
		    FileSystem fs = file.getFileSystem(conf);
		    final FSDataInputStream fileIn = fs.open(fsplit.getPath());
		    fileIn.seek(fsplit.getStart());
		    String line = fileIn.readLine();
		    //final ObjectInputStream object = new ObjectInputStream(fileIn);
		   //FPeopleWritable2 fp = new FPeopleWritable2(peopleNum/mmn,geneLength);
		   //fp.readFields(fileIn);
		    LOG.info("XXX: " + fsplit.getStart() + "line is " + line);

			return new DRecordReader(Integer.parseInt(line),dmn);
		}
	}

	public static class DRecordReader implements RecordReader<IntWritable,IntWritable> {
		private int total = 0;
		private int seg = 0;
		boolean mark = true;
		
		
		public DRecordReader(int seg, int all) {
			this.seg = seg;
			total = all;
		}
		
		public boolean next(IntWritable key, IntWritable value)
		throws IOException {

			
			key.set(seg);
			value.set(total);
			

			if (mark) {
				mark = false;
				return true;
			} else
				return false;
		}
		
		public IntWritable createKey() { return new IntWritable(); }
		
		public IntWritable createValue() { return new IntWritable(); }
		
		public long getPos() throws IOException { return 1; }
		
		public void close() throws IOException { }
		
		public float getProgress() throws IOException {
			return 1;
		}

	}

	public static class DMap extends MapReduceBase implements
	Mapper<IntWritable, IntWritable, DoubleWritable, Text > {
		
		public void map(IntWritable key, IntWritable value,
				OutputCollector<DoubleWritable, Text> output, Reporter reporter)
		throws IOException {

			//String res = String.valueOf(PassWord.passWord());
			//String res = GA3.ga(geneLength, dpn, p_cross, p_mut, i_num,dfc,dmag);
			String res = "test";
			if (res == "") {
				res = "no value";
			}

			Text rt = new Text(res);
			DoubleWritable keyValue = new DoubleWritable(key.get()+value.get());
			output.collect(keyValue, rt);
			reporter.setStatus("key= " + key.get() + " " + value.get());
		}
	}

	public static class DReduce extends MapReduceBase implements
	Reducer<DoubleWritable, Text, DoubleWritable, Text> {
		private static final Log LOG = LogFactory.getLog(DReduce.class);
		public void reduce(DoubleWritable key, Iterator<Text> values,
				OutputCollector<DoubleWritable, Text> output, Reporter reporter)
		throws IOException {

			output.collect(key, new Text(values.next()));
		}
	}
	
	
/*	public static String ga(int geneLength,
			int peopleNum, double p_cross,
			double p_mut, long di_num,Fc3 fc,
			double start, double all) {

		String value = "";
		String time = "";
		String it = "";
		double g_b = Double.NEGATIVE_INFINITY;

		long start_time = System.currentTimeMillis();
		FPeople3 my_p = new FPeople3(peopleNum,geneLength);
		FEvolution3 my_e = new FEvolution3(0.1, p_cross, p_mut);
		fc.caculate(my_p.persons, start, all);

		for (int i=0; i < di_num; i++) {

			if (i%ri == 0 || i == i_num - 1){
				value = value + -my_p.getBest().fitness + "\t";
				time = time + (System.currentTimeMillis() - start_time) + "\t";
				it = it + i + "\t";
			}
		

			my_e.select(my_p);
			my_e.crossOver(my_p);
			my_e.mutation(my_p);
			fc.caculate(my_p.persons, start, all);
		}
		
		return value + "\n" + time + "\n" + it + "\n";
	}*/
	
//	public static class PassWord {
	/*public static int p[] = {1,2,3,4};
		public static int pn = 1;
		public static int pl = 4;
		public static int id[][] = new int[pn][pl];
		
		public static void init(){
			for (int i = 0; i < pn; i++) {
				for (int j = 0; j < pl; j++) {
					Random r =new Random();
					id[i][j] = r.nextInt(pl+1);
				}
			}
		}
		
	
		public static int passWord(){
			int it=0;
			
			init();
			while (!Arrays.equals(id[0], p)) {
				int pos = new Random().nextInt(pl);
				id[0][pos] = new Random().nextInt(pl+1);
				
				System.out.print("password " );
				for (int x: id[0] ){
					System.out.print(x);	
				}
				System.out.print("\n");	
				it++;
			}
			System.out.println("iteration number is: " + it);
			return it;
		}*/
		
//		public static void main(String args[]) {
//			
//			int mit = 0;
//			
//			for (int i = 0; i < 1000; i++) {
//				
//				mit = mit + passWord();
//			}
//			
//			System.out.println("iteration number is: " + mit);
//			
//			System.out.println("leave main");
//		}
		
//	}

	public static void democratic() {
		//删除hadoop的运行目录,否则hadoop会提示错误

		String cmd = hp + "/bin/hadoop fs -rmr " + dout;
		RWF4.execmd(cmd);

		//set hadoop task
		JobConf conf = new JobConf(D4.class);
		conf.setJobName("D4");
		
		conf.set("mapred.child.java.opts","-Xmx1000m");
		conf.set("mapred.reduce.max.attempts", "100");
		conf.set("mapred.task.timeout", "0");

		//control map's number
		conf.setNumReduceTasks(1);
		
		conf.setOutputKeyClass(DoubleWritable.class);
		conf.setOutputValueClass(Text.class);

		conf.setMapperClass(DMap.class);

		conf.setReducerClass(DReduce.class);

		conf.setInputFormat(DPGAInputFormat.class);
		//conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);


		FileOutputFormat.setOutputPath(conf, new Path(dout));
		FileInputFormat.setInputPaths(conf, new Path(din));

		long t1 = System.currentTimeMillis();
		//运行测试内容
		try {
			JobClient.runJob(conf);
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
		}
		long t2 = System.currentTimeMillis();

		LOG.info(t2-t1);

		
	}
	

	public static void main(String[] args) throws Exception {
		LOG.info("f Enter main");

		
		for (int i=0; args != null && i < args.length; i++) {
			if (args[i].equalsIgnoreCase("-mn") || args[i].equalsIgnoreCase("-map_num")) {
				i++;
				dmn = Integer.parseInt(args[i]);
				System.out.println("map number is: " + args[i]);
			} else if (args[i].equalsIgnoreCase("-ln") ||args[i].equalsIgnoreCase("-loop_num")) {
				i++;
				dln = Integer.parseInt(args[i]);
				System.out.println("loop number is: " + args[i]);
			} else {
				System.out.println("Usage: -mn[map_num] 10 -ln[loop_num] 25");
				System.exit(-1);
			}
        }   
		//reserve results
		String cmd = "mkdir -p ./res";
		RWF4.execmd(cmd);
		
		//for running much programmers
		String mark = RWF4.mkMark("./res/dmark", "pw");
		mark = mark + "_" + dln + "_" + dmn;

		//put different results of different program to different directory 
		cmd = "mkdir -p " + drp;
		RWF4.execmd(cmd);
		
		dout = dout + "/" + mark;
		din = din + "/" + mark;
		
	
		String seg = "";
		for (int i = 0; i < dmn; i++ ) {
			seg = seg + i + "\n";
		}
		System.out.print("length is " + seg.length() + seg);
		System.out.println("end");
		
		RWF4.whf(seg, din + "/seg");

		String rt = "";
		//循环多次
		for (int i = 0; i < dln; i++) {
			//dfc.init();
			long t = System.currentTimeMillis();
			//启动并行
			democratic();
			RWF4.wlf(RWF4.rhf(dout + "/part-00000"), drp + "/d_r" + mark);
			rt = rt + (System.currentTimeMillis() - t) + "\t";
			System.out.println(i + " Game over");
		}
		rt = rt + "\n";
		RWF4.wlf(rt.getBytes(), drp + "/dt" + mark);
		//求出最好、最差、平均解 
		LOG.info("f Leave main");
	}

}





